﻿using Mina.Core.Buffer;
using Mina.Core.Session;
using Mina.Transport.Socket;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace HttpPerfLoad
{
    class ConnectionManager
    {
        public int numConnections;
        public event EventHandler<TickInfo> StatisticsTick;
        public event EventHandler<ConnectionError> ConnectError;
        public event EventHandler<ProgressInfo> ConnectionProgress;
        public event EventHandler AllConnectionCompleted;
        public event EventHandler AllConnectionClosed;
        public string ip;
        public int port;
        internal bool stopRequested;
        public ConnectionManager(string ip, int port, int numConnections)
        {
            this.numConnections = numConnections;
            this.ip = ip;
            this.port = port;
        }

        public void Start()
        {
            new Thread(ThreadMain).Start();
            this.responseData = "";
        }

        public void Stop()
        {
            if (!stopRequested)
            {

                stopRequested = true;
                var sessions = GetAliveSessions();
                foreach (var item in sessions)
                {
                    item.Session?.Close(true);
                    item.connector.Dispose();
                    item.SessionBroken = true;
                }
            }
        }
        List<Pair> aliveSessions = new List<Pair>();
        public List<Pair> GetAliveSessions()
        {
            lock (aliveSessions)
            {
                return aliveSessions.ToList();
            }
        }
        public string responseData = "";
        public string errorText = "";
        //public static int numAliveSessions;
        static System.Threading.Timer timer;

        public void ThreadMain()
        {
            int workers;
            int completionPortThreads;
            ThreadPool.GetMaxThreads(out workers, out completionPortThreads);
            Console.WriteLine($"ThreadPool workers: {workers},completionPortThreads: {completionPortThreads}");

            timer = new System.Threading.Timer(TimerCallback, null, 0, 50);
            for (int i = 0; i < (int)this.numConnections; i++)
            {
                // Create TCP/IP connector.
                AsyncSocketConnector connector = new AsyncSocketConnector();
                connector.ExceptionCaught += (s, e) => errorText = e.Exception.Message;
                connector.ReuseBuffer = true;
                //connector.FilterChain.AddLast("codec", new ProtocolCodecFilter(new HttpCodecFactory(HttpCodecRole.HttpClient, "http://192.1668.199.187:9000/")));
                // Set connect timeout.

                //connector.ConnectTimeoutInMillis = 120 * 1000L;
                // Set reader idle time to 10 seconds.
                // sessionIdle(...) method will be invoked when no data is read
                // for 10 seconds.
                connector.SessionOpened += (s, e) =>
                {
                    //Interlocked.Increment(ref numAliveSessions);
                    if (this.stopRequested)
                    {
                        e.Session.Close(true);

                    }
                };

                // Print out total number of bytes read from the remote peer.
                connector.SessionClosed += (s, e) =>
                {
                    lock (this.aliveSessions)
                    {
                        var pairClosed = e.Session.GetAttribute("pair") as Pair;
                        pairClosed.connector.Dispose();
                        this.aliveSessions.Remove(pairClosed);
                    }
                    //Interlocked.Decrement(ref numAliveSessions);
                };

                connector.SessionIdle += (s, e) =>
                {
                    //if (e.IdleStatus == IdleStatus.ReaderIdle)
                    //e.Session.Close(true);
                };
                //connector.MessageSent += (s, e) =>
                //{
                //    Console.WriteLine(e.Message as string);
                //};
                connector.MessageReceived += (s, e) =>
                {
                    var buff = (e.Message as IoBuffer);
                    var bytes = new byte[buff.Remaining];
                    buff.Get(bytes, 0, buff.Remaining);
                    responseData = Encoding.UTF8.GetString(bytes);
                };

                var pair = new Pair(aliveSessions, this, connector, ip, port);
                if (!pair.ConnectInBackground(i))
                {
                    Console.WriteLine("error on quene background work");
                    break;
                }
                else
                {
                    Thread.Sleep(10);
                }
                if (stopRequested)
                {
                    break;
                }
            }
        }
        internal void OnConnectError(string errorText, int index)
        {
            this.errorText = errorText;
            if (this.ConnectError != null)
            {
                this.ConnectError.Invoke(this, new ConnectionError { ErrorText = errorText, ConnectionIndex = index });
            }
        }
        public int numProcessed;
        internal void OnConnectionProgress(ProgressInfo progressInfo)
        {
            if (this.ConnectionProgress != null)
            {
                this.ConnectionProgress.Invoke(this, progressInfo);
            }
            numProcessed = progressInfo.current;
            if (numProcessed == progressInfo.total)
            {
                if (AllConnectionCompleted != null)
                    AllConnectionCompleted.Invoke(this, EventArgs.Empty);
            }
        }

        void TimerCallback(object state)
        {
            if (StatisticsTick != null)
            {
                var aliveSessionList = GetAliveSessions();
                var totalRecv = aliveSessionList.Sum(x => x.connector.Statistics.ReadBytes);
                var totalSend = aliveSessionList.Sum(x => x.connector.Statistics.WrittenBytes);

                var speedRecv = aliveSessionList.Sum(x => (int)x.connector.Statistics.ReadBytesThroughput);
                var speedSend = aliveSessionList.Sum(x => (int)x.connector.Statistics.WrittenBytesThroughput);


                StatisticsTick.Invoke(this, new TickInfo
                {
                    aliveSessions = aliveSessionList.Count,
                    totalRead = totalRecv,
                    totalWrite = totalSend,
                    avgRead = speedRecv,
                    avgWrite = speedSend,
                });
                if (aliveSessionList.Count == 0 && stopRequested)
                {
                    timer.Dispose();
                    if (this.AllConnectionClosed != null) {
                        this.AllConnectionClosed.Invoke(this, null);
                    }
                }
                //Console.Title = $"HttpPerfLoad | threads: {numThread}, connection: {copiedList.Count}, R: {(totalRecvBytes / 1024.0).ToString("f2")}MB, W: {(totalSendBytes / 1024.0).ToString("f2")}MB, SR_AVG: {speedRecv.ToString("f2")} bytes/s, SW_AVG: {speedSend.ToString("f2")}bytes/s";
            }
        }
    }
}
